/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.antgroup.geaflow.memory;

public class HeapMemoryPool extends AbstractMemoryPool<byte[]> {

    public HeapMemoryPool(MemoryManager memoryManager, int pageSize, int maxOrder, int pageShifts,
                          int chunkSize) {
        super(memoryManager, pageSize, maxOrder, pageShifts, chunkSize);
    }

    @Override
    protected void allocateBytebuf(ByteBuf<byte[]> buf, int normCapacity, MemoryGroup group) {

        group.allocate(normCapacity, getMemoryMode());
        if (q050.allocate(buf, normCapacity) || q025.allocate(buf, normCapacity) || q000
            .allocate(buf, normCapacity) || qInit.allocate(buf, normCapacity) || q075
            .allocate(buf, normCapacity)) {
            return;
        }

        Chunk<byte[]> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
        long handle = c.allocate(normCapacity);
        assert handle > 0;
        c.initBuf(buf, handle, normCapacity);
        qInit.add(c);
        currentChunkNum ++;
        updateAllocateMemory();
        reloadMemoryStatics(getMemoryMode());
    }

    @Override
    MemoryMode getMemoryMode() {
        return MemoryMode.ON_HEAP;
    }

    @Override
    boolean canExpandCapacity() {
        return true;
    }

    @Override
    void shrinkCapacity() {

    }

    @Override
    protected String dump() {
        return null;
    }

    @Override
    void destroyChunk(Chunk<byte[]> chunk) {
        //wait gc
        currentChunkNum --;
        updateAllocateMemory();
        reloadMemoryStatics(getMemoryMode());
    }

    @Override
    protected ByteBuf<byte[]> newByteBuf(int size) {
        return new ByteBuf<>();
    }

    @Override
    protected Chunk<byte[]> newChunk(int pageSize, int maxOrder, int pageShifts, int chunkSize) {
        return new Chunk<>(this, new byte[chunkSize], pageSize, maxOrder, pageShifts,
            chunkSize, 0);
    }
}
